package dts_tcp_client

import (
	"context"
	"encoding/binary"
	"fmt"
	"github.com/samber/lo"
	"github.com/sirupsen/logrus"
	"github.com/zing-dev/general-protocol-lib/log"
	"github.com/zing-dev/go-bit-bytes/bit"
	"io"
	"net"
	"sync/atomic"
	"time"
)

func (c *Client) Run() {
	// 正在连接
	c.setStatus(Connecting)

	c.pushMessage(I, fmt.Sprintf("当前客户端<%s>连接DTS服务端<%s>...", c.info.Name, c.config.Server))

	conn, err := net.DialTimeout("tcp", c.addr.String(), c.config.Timeout)
	if err != nil {
		c.setStatus(NotConnected)
		c.pushMessage(E, fmt.Sprintf("当前客户端<%s>连接DTS服务端<%s>失败: %s", c.info.Name, c.config.Server, err))
		if c.config.Reconnect {
			go c.Run()
		}
		return
	}

	c.conn = conn.(*net.TCPConn)

	err = c.conn.SetReadBuffer(1024 * 1024 * 2)
	if err != nil {
		c.setStatus(Disconnected)
		if c.config.Reconnect {
			go c.Run()
		}
		return
	}

	c.pushMessage(I, fmt.Sprintf("当前客户端<%s>连接DTS服务端<%s>成功", c.info.Name, c.config.Server))

	c.setStatus(Connected)

	c.wait.Add(1)
	go c.handler()
}

func (c *Client) handler() {
	c.ctx, c.cancel = context.WithCancel(c.parentContext)
	var stop bool
	defer func() {
		c.pushMessage(W, "处理消息协程退出...")
		c.wait.Done()
		err := c.conn.Close()
		if err != nil {
			c.pushMessage(E, fmt.Sprintf("关闭客户端失败: %s", err))
		}
		if !stop {
			if c.config.Reconnect {
				c.setStatus(Disconnected)
				go c.Run()
				return
			}
			c.Close()
		}
	}()
	bit.DefaultEndPont = binary.LittleEndian
	for {
		select {
		case <-c.ctx.Done():
			stop = true
			return
		default:
			header := make([]byte, HeaderLength)
			n, err := io.ReadFull(c.conn, header)
			if err != nil {
				c.pushMessage(E, fmt.Sprintf("读取头信息失败: %s", err))
				return
			}
			if n != len(header) {
				c.pushMessage(E, "读取头信息长度非法")
				return
			}

			// 获取当前包的总大小
			total := bit.ToUint16(header[9:11])

			// 余下数据包括包尾
			data := make([]byte, total-HeaderLength)

			n, err = io.ReadFull(c.conn, data)
			if err != nil {
				c.pushMessage(E, fmt.Sprintf("读取数据信息失败: %s", err))
				return
			}
			if n != len(data) {
				c.pushMessage(E, "读取数据信息长度非法")
				return
			}
			if data[len(data)-1] != DataFrameEnd {
				c.pushMessage(E, "非法帧尾数据")
				continue
			}

			switch header[0] {
			case DataFrameTemperatureStart:
				flag := DataFlag(data[0])
				if c.config.FilterSignal && flag != FlagTemperature {
					// 过滤温度斯托克和反斯托克信号数据
					// FlagStokes
					// FlagAntiStokes
					continue
				}
				_ = data[1:2]                    // 1~2 设备id
				_ = bit.ToInt16(data[2:4])       // 2~3 通道标志位
				start := bit.ToUint16(data[4:6]) // 4~5 数据起点位置
				end := bit.ToUint16(data[6:8])   // 6~7 数据终点位置
				count := bit.ToInt16(data[8:10]) // 8~9 总通道数
				current := data[10] + 1          // 10 当前通道号,获取的通道号从0开始,显示+1

				var (
					length       = (len(data) - 2 - 2 - 2 - 2 - 2 - 1 - 1) / 2
					temperatures = make([]float32, length)
					info         = Info{Id: c.info.Id, Name: c.info.Name, DeviceNumber: header[8]}
				)

				for i := range temperatures {
					temperatures[i] = float32(bit.ToInt16(data[11+i*2:i*2+13])-20000) / float32(100)
				}
				if c.config.ChannelMapping != nil && c.config.ChannelMapping[current] != 0 {
					info.ChannelId = c.config.ChannelMapping[current]
				}
				channelTemperature := ChannelTemperature{
					Info:          info,
					Flag:          flag,
					ChannelNumber: current,
					ChannelCount:  byte(count),
					Start:         float32(start) * c.config.Precision,
					End:           float32(end) * c.config.Precision,
					Temperatures:  temperatures,
					At:            time.Now(),
				}
				c.locker.RLock()
				lo.ForEach(lo.Values(c.callChannelTemperature), func(call CallChannelTemperature, _ int) { go call(channelTemperature) })
				c.locker.RUnlock()
				c.pushMessage(I, fmt.Sprintf("设备<%s>通道<%d>在<%.3f>~<%.3f>米温度更新", channelTemperature.Info.Name, channelTemperature.ChannelNumber, channelTemperature.Start, channelTemperature.End))
			case DataFrameFiberBrokenAlarmStart:
				var (
					position      = bit.ToUint16(data[1:3])
					channelNumber = data[0] + 1
					info          = Info{Id: c.info.Id, Name: c.info.Name, DeviceNumber: header[8]}
				)
				if c.config.ChannelMapping != nil && c.config.ChannelMapping[channelNumber] != 0 {
					info.ChannelId = c.config.ChannelMapping[channelNumber]
				}
				fiberBroken := FiberBroken{
					Info:          info,
					ChannelNumber: channelNumber,
					Position:      float32(position) * c.config.Precision,
					At:            time.Now(),
				}
				c.locker.RLock()
				lo.ForEach(lo.Values(c.callFiberBroken), func(call CallFiberBroken, _ int) { go call(fiberBroken) })
				c.locker.RUnlock()
				c.pushMessage(W, fmt.Sprintf("设备<%s>通道<%d>在<%f>米发生断纤报警", fiberBroken.Info.Name, fiberBroken.ChannelNumber, fiberBroken.Position))
			case DataFrameConstantTemperatureAlarmStart, DataFrameDifferentialTemperatureAlarmStart:
				c.HandlerAlarm(header, data)
			case DataFrameYesStart:
				c.pushMessage(W, "参数校验Yes")
			case DataFrameNoStart:
				c.pushMessage(W, "参数校验No")
			default:
				c.pushMessage(W, fmt.Sprintf("其他数据包包协议: %X", header))
			}
		}
	}
}

func (c *Client) HandlerAlarm(header, data []byte) {
	var (
		count   = data[0]     // 通道报警总计数
		current = data[1] + 1 // 当前报警通道号
		alarms  = make([]Alarm, count)
	)
	for i := range alarms {
		alarms[i] = Alarm{
			AreaNumber:  int16(data[i*5+2]) + 1,                       // 区域号+1
			Position:    float32(bit.ToUint16(data[i*5+3:i*5+5])) / 2, // 报警位置精度0.5M
			Temperature: float32(bit.ToUint16(data[i*5+5 : i*5+7])),   // 报警温度精度0.1℃
		}
	}
	info := Info{Id: c.info.Id, Name: c.info.Name, DeviceNumber: header[8]}
	if c.config.ChannelMapping != nil && c.config.ChannelMapping[current] != 0 {
		info.ChannelId = c.config.ChannelMapping[current]
	}
	temperatureAlarm := TemperatureAlarm{Info: info, ChannelNumber: current, Alarms: alarms, At: time.Now()}
	if header[0] == DataFrameConstantTemperatureAlarmStart {
		temperatureAlarm.Type = ConstantTemperature
		c.pushMessage(W, fmt.Sprintf("设备<%s>通道<%d>发生定温报警", temperatureAlarm.Info.Name, temperatureAlarm.ChannelNumber))
	} else if header[0] == DataFrameDifferentialTemperatureAlarmStart {
		temperatureAlarm.Type = DifferentialTemperature
		c.pushMessage(W, fmt.Sprintf("设备<%s>通道<%d>发生温差报警", temperatureAlarm.Info.Name, temperatureAlarm.ChannelNumber))
	} else {
		return
	}
	c.locker.RLock()
	lo.ForEach(lo.Values(c.callTemperatureAlarm), func(call CallTemperatureAlarm, _ int) { go call(temperatureAlarm) })
	c.locker.RUnlock()
}

func (c *Client) setStatus(status Status) {
	atomic.StoreInt32(&c.status, int32(status))
	lo.ForEach(lo.Values(c.callConnectChange), func(call CallConnectChange, _ int) { go call(status) })
}

func (c *Client) GetStatus() Status {
	status := atomic.LoadInt32(&c.status)
	return Status(status)
}

func (c *Client) IsConnected() bool {
	return c.GetStatus() == Connected
}

func (c *Client) pushMessage(level logrus.Level, message string) {
	if level >= I && !c.config.Debug {
		return
	}
	text, err := level.MarshalText()
	if err != nil {
		return
	}
	c.locker.RLock()
	lo.ForEach(lo.Values(c.callMessage), func(call CallMessage, _ int) {
		go call(Message{Level: string(text), Message: message, At: time.Now()})
	})
	c.locker.RUnlock()
}

func (c *Client) Close() {
	if !c.IsConnected() {
		return
	}
	err := c.conn.Close()
	if err != nil {
		c.pushMessage(I, fmt.Sprintf("当前客户端<%s>关闭连接失败:%s", c.info.Name, c.config.Server))
	}
	c.cancel()
	c.wait.Wait()
	c.pushMessage(I, fmt.Sprintf("当前客户端<%s>断开DTS服务端<%s>成功", c.info.Name, c.config.Server))
	c.setStatus(Disconnected)
}

func (c *Client) TestRun() {
	c.setStatus(Connecting)
	c.ctx, c.cancel = context.WithCancel(context.Background())
	c.wait.Add(1)
	go func() {
		defer c.wait.Done()
		c.setStatus(Connected)
		for {
			select {
			case <-c.ctx.Done():
				log.L.Warn("return...")
				c.setStatus(Disconnected)
				return
			default:
				log.L.Info("default...")
				time.Sleep(time.Second)
			}
		}
	}()
}
func (c *Client) TestStop() {
	c.cancel()
	c.wait.Wait()
}
